package com.shujia.flink.tf

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow

object Demo7Window {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)


    val kvDS: DataStream[(String, Int)] = linesDS.flatMap(_.split(",")).map((_, 1))


    val keyBeyDs: KeyedStream[(String, Int), String] = kvDS.keyBy(_._1)

    /**
      * 每一个5秒统计单的数量,统计最近15的数据
      *
      */

    //5秒一个窗口
    val windowDS: WindowedStream[(String, Int), String, TimeWindow] = keyBeyDs
      //      .timeWindow(Time.seconds(5)) //滚动窗口（两个窗口没有交接）
      .timeWindow(Time.seconds(15), Time.seconds(5)) // 滑动窗口

    //在窗口内进行聚合计算
    val reduceDS: DataStream[(String, Int)] = windowDS.reduce((x, y) => (x._1, x._2 + y._2))

    reduceDS.print()


    env.execute()


  }

}
